package com.tmsb.utils;

import com.alibaba.fastjson.JSON;

import com.tmsb.pojo.ESAlarm;
import com.tmsb.pojo.ESStockAlarm;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;
import java.util.*;

/**
 * @Author wuxiangwu
 * @Date 2020/8/12 13:33
 * @Description ES Stock的报警逻辑处理
 * @Version 1.0
 */
public  class AlarmProcessWindowFunction  extends ProcessWindowFunction<ESStockAlarm, String, Tuple, TimeWindow> {
        // 用于记录每个用户上一时刻的balance状态值
        private ValueState<BigDecimal> balanceState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            balanceState = getRuntimeContext().getState(new ValueStateDescriptor<BigDecimal>("balanceState", BigDecimal.class));
        }

        @Override
        public void close() throws Exception {
            super.close();
        }

    @Override
    public void process(Tuple tuple, Context context, Iterable<ESStockAlarm> elements, Collector<String> out) throws Exception {
        BigDecimal lastBalance;
        BigDecimal amount;
        Tuple2<Long, String> tuple2 = (Tuple2<Long, String>) tuple;

        // 获取出窗口中的所有数据
        Iterator<ESStockAlarm> iterator = elements.iterator();
        BigDecimal totalAmount = new BigDecimal(0);
        // 判断是否存有上一个状态值，如果没有就判断这个窗口的的数值是不是同号
        //如果同为正就取窗口内的最大balance，如果同为负就取窗口内的最小balance
        Double temp = null;
        if (balanceState.value() == null) {
            List<BigDecimal> balancesList = new ArrayList<>();
            while (iterator.hasNext()) {
                ESStockAlarm next = iterator.next();
                amount = next.getAmount();
                totalAmount = totalAmount.add(amount);
                balancesList.add(next.getBalance());
                if (temp == null) {
                    temp = amount.doubleValue();
                } else if (temp * amount.doubleValue() < 0) {
                    return;
                }
            }
            balanceState.update(temp > 0 ? Collections.max(balancesList) : Collections.min(balancesList));
        } else {
            List<Double> balancesList = new ArrayList<>();
            //TODO 定义一个map，用来存放这个窗口的last和balance值，以便取出这个窗口last最大的balance
             HashMap<Long, Double> map = new HashMap<>();
            //TODO 如果保存了上一次的状态，直接取出balance值
            lastBalance = balanceState.value();
            //TODO 遍历累加出这一秒的amount总和
            while (iterator.hasNext()) {
                ESStockAlarm next = iterator.next();
                amount = next.getAmount();
                balancesList.add(next.getBalance().doubleValue());
                totalAmount = totalAmount.add(amount);
                map.put(next.getLast(),next.getBalance().doubleValue());
            }
            //TODO 将这一秒的amount总和加上lastBalance之后更新到状态
            balanceState.update(lastBalance.add(totalAmount));
            //TODO 校验上一秒的balance加上amount是否在集合的balance里面
            Double balance = balanceState.value().doubleValue();
            if (balancesList.contains(balance)) {
      //          System.out.println("用户id为：" + tuple2.f0 + "的状态正常, name = " + tuple2.f1 + ",balance=" + balance + ",create_time=" + context.window().getEnd() / 1000+",last_balance="+lastBalance);
            } else {
                //TODO 如果算出来的balance不在balanceList里面，就去计算它与这个窗口最后一个last对应的balance值的差值
                Set<Long> lastList = map.keySet();
                Long maxLast = Collections.max(lastList);
                Double getValue = map.get(maxLast);
                String diff = MyMath.sub(balance.toString(), getValue.toString());
                // TODO 封装异常数据输出到下游
                ESAlarm ea = new ESAlarm();
                ea.setId(Util.getAlarmId());
                ea.setExp_value(balanceState.value());
                ea.setGet_value(getValue);
                ea.setWarn_title(tuple2.f1);
                ea.setDiff(diff);
                ea.setWarn_detail("user_id="+tuple2.f0+",balance="+balance+",create_time=" + context.window().getEnd()/1000+",last="+context.window().getEnd());
                out.collect(JSON.toJSONString(ea));
       //         System.out.println("用户id为：" + tuple2.f0 + "的记录异常, name = " + tuple2.f1 + ",balance=" + balance + ",create_time=" + context.window().getEnd() / 1000+",last_balance="+lastBalance);
//                for (Double bigDecimal : balancesList) {
//                    System.out.println("balance = "+bigDecimal+",windowEnd = "+context.window().getEnd());
//                }
                balanceState.clear();
            }

        }
//            ESAlarm sa = new ESAlarm();
//            sa.setAmount(totalAmount);
//            sa.setUser_id(tuple2.f0);
//            sa.setBalance(balanceState.value());
//            sa.setName(tuple2.f1);
//            sa.setCreated_time(context.window().getEnd() / 1000);
        //TODO 输入这5分钟用户的balance到下游
    }
}
